package driver

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"strconv"
	"strings"
	"time"

	"task/config"
	"task/pb/business"
	"task/pb/dispatch"

	"github.com/apache/pulsar-client-go/pulsar"
	"github.com/bwmarrin/snowflake"
	"github.com/go-redis/redis/v8"
	_ "github.com/go-sql-driver/mysql"
	"github.com/go-xorm/xorm"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	log "github.com/sirupsen/logrus"
	"google.golang.org/grpc"
)

var (
	Engine            *xorm.Engine
	redisClient       *redis.Client
	node              *snowflake.Node
	PulsarClient      pulsar.Client
	DispatchProducer  pulsar.Producer
	ExecuteProducer   pulsar.Producer
	Business1Producer pulsar.Producer
	Business2Producer pulsar.Producer
	CallbackProducer  pulsar.Producer
	DispatchClient    dispatch.DispatchServiceClient
	BusinessClient    business.BusinessServiceClient
)

// InitEngine init engine.
func InitEngine() {
	var err error
	dsn := fmt.Sprintf("%s:%s@(%s:%d)/%s?charset=utf8mb4",
		config.Conf.Engine.User, config.Conf.Engine.Passwd,
		config.Conf.Engine.Host, config.Conf.Engine.Port, config.Conf.Engine.DB)
	if Engine, err = xorm.NewEngine("mysql", dsn); err != nil {
		log.Panic("Failed to create Engine:", err)
	}
	if err = Engine.Ping(); err != nil {
		log.Panic("Failed to ping Engine:", err)
	}
	log.Info("Init Engine success")
	return
}

// InitRedis init redis.
func InitRedis() {
	var (
		ctx    = context.Background()
		result string
		err    error
	)
	redisHost := "redis"
	redisPort := "6379"
	passwd := "Server.Sues.112"
	redisClient = redis.NewClient(&redis.Options{
		Addr:     fmt.Sprintf("%s:%s", redisHost, redisPort),
		Password: passwd,
		DB:       0,
	})
	if result, err = redisClient.Ping(ctx).Result(); err != nil {
		log.Panic(err)
	}
	log.Infof("redis ping result: %s", result)
	log.Info("init redis success...")
	return
}

// InitSnowNode init snow node.
func InitSnowNode() {
	var (
		ip     string
		nodeId int64
		err    error
	)
	// 获取ip
	addrList, _ := net.InterfaceAddrs()
	for _, addr := range addrList {
		if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
			if ipnet.IP.To4() != nil {
				ip = ipnet.IP.String()
				break
			}
		}
	}
	log.Infof("Get ip(%s)", ip)
	// ip取最后一段作为nodeId
	tmp := strings.Split(ip, ".")
	if nodeId, err = strconv.ParseInt(tmp[len(tmp)-1], 10, 64); err != nil {
		log.Panic("Ip is wrong:", err)
	}
	log.Infof("Snow node id(%d)", nodeId)
	if node, err = snowflake.NewNode(nodeId); err != nil {
		log.Panic("Failed to create snow node:", err)
	}
	log.Info("Init snow node success")
	return
}

// GetSnowFlakeId get snow id.
func GetSnowFlakeId() int64 {
	return node.Generate().Int64()
}

// InitPulsarClient init pulsar client.
func InitPulsarClient() {
	var err error
	if PulsarClient, err = pulsar.NewClient(pulsar.ClientOptions{
		URL:               config.Conf.Pulsar.Address,
		OperationTimeout:  time.Duration(config.Conf.Pulsar.Timeout) * time.Second,
		ConnectionTimeout: time.Duration(config.Conf.Pulsar.Timeout) * time.Second,
	}); err != nil {
		log.Panic("Failed to create pulsar client:", err)
	}
	return
}

// InitDispatchProducer init dispatch producer.
func InitDispatchProducer() {
	var err error
	if DispatchProducer, err = PulsarClient.CreateProducer(pulsar.ProducerOptions{
		Topic: config.Conf.Pulsar.DispatchTopic,
	}); err != nil {
		log.Panic("Failed to create dispatch producer:", err)
	}
	return
}

// InitExecuteProducer init execute producer.
func InitExecuteProducer() {
	var err error
	if ExecuteProducer, err = PulsarClient.CreateProducer(pulsar.ProducerOptions{
		Topic: config.Conf.Pulsar.ExecuteTopic,
	}); err != nil {
		log.Panic("Failed create execute producer:", err)
	}
	return
}

// InitBusiness1Producer init business1 producer.
func InitBusiness1Producer() {
	var err error
	if Business1Producer, err = PulsarClient.CreateProducer(pulsar.ProducerOptions{
		Topic: config.Conf.Pulsar.Business1Topic,
	}); err != nil {
		log.Panic("Failed to create business1 producer:", err)
	}
	return
}

// InitBusiness2Producer init business2 producer.
func InitBusiness2Producer() {
	var err error
	if Business2Producer, err = PulsarClient.CreateProducer(pulsar.ProducerOptions{
		Topic: config.Conf.Pulsar.Business2Topic,
	}); err != nil {
		log.Panic("Failed to create business2 producer:", err)
	}
	return
}

// InitCallbackProducer init callback producer.
func InitCallbackProducer() {
	var err error
	if CallbackProducer, err = PulsarClient.CreateProducer(pulsar.ProducerOptions{
		Topic: config.Conf.Pulsar.CallbackTopic,
	}); err != nil {
		log.Panic("Failed to create callback producer:", err)
	}
	return
}

// StartMetricsServer start metrics server.
func StartMetricsServer() {
	http.Handle("/metrics", promhttp.Handler())
	if err := http.ListenAndServe(config.Conf.Metrics.Address, nil); err != nil {
		log.Panic("Failed to start metrics server:", err)
	}
}

// InitDispatchClient init dispatch client.
func InitDispatchClient() *grpc.ClientConn {
	var err error
	conn := new(grpc.ClientConn)
	addr := fmt.Sprintf("%s:%d", config.Conf.TaskDispatch.Name, config.Conf.TaskDispatch.GrpcPort)
	if conn, err = grpc.Dial(addr, grpc.WithInsecure()); err != nil {
		log.Panic("Failed to create task dispatch grpc client:", err)
	}
	DispatchClient = dispatch.NewDispatchServiceClient(conn)
	return conn
}

// InitBusinessClient init business client.
func InitBusinessClient() *grpc.ClientConn {
	var err error
	conn := new(grpc.ClientConn)
	addr := fmt.Sprintf("%s:%d", config.Conf.Business.Name, config.Conf.Business.GrpcPort)
	if conn, err = grpc.Dial(addr, grpc.WithInsecure()); err != nil {
		log.Panic("Failed to create business grpc client:", err)
	}
	BusinessClient = business.NewBusinessServiceClient(conn)
	return conn
}
